package com.shanks.framework.mq.rocketmq.config;

import com.shanks.framework.mq.common.constant.MQConstant;
import com.shanks.framework.mq.domain.Subscription;
import com.shanks.framework.mq.enums.MQZone;
import com.shanks.framework.mq.exception.MQException;
import com.shanks.framework.mq.processor.MQProcessor;
import com.shanks.framework.mq.rocketmq.consumer.RocketMQConsumerAttribute;
import com.shanks.framework.mq.rocketmq.listener.ConcurrentMessageListener;
import jakarta.annotation.Resource;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.env.Environment;

import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

public class RocketMQConsumerInitializer {
    private static final Logger logger = LoggerFactory.getLogger(RocketMQConsumerInitializer.class);

    private Map<String, DefaultMQPushConsumer> consumersMap = new ConcurrentHashMap<>();

    private Map<String, RocketMQConsumerAttribute> consumerAttributeMap = new ConcurrentHashMap<>();

    @Resource
    private Environment environment;

    private MQProcessor ackProcessor;


    public RocketMQConsumerInitializer(MQProcessor ackProcessor) {
        this.ackProcessor = ackProcessor;
    }

    public Map<String, DefaultMQPushConsumer> getConsumersMap() {
        return consumersMap;
    }

    public DefaultMQPushConsumer createAndStartMQConsumer(RocketMQConsumerAttribute rocketMQConsumerAttribute) throws MQException {
        try {
            // 创建消费者
            DefaultMQPushConsumer mqConsumer = createAndInitMQConsumer(rocketMQConsumerAttribute);
            consumersMap.putIfAbsent(mqConsumer.getConsumerGroup(), mqConsumer);
            consumerAttributeMap.putIfAbsent(rocketMQConsumerAttribute.getGroupName(), rocketMQConsumerAttribute);

            logger.info("RocketMQ consumer started successfully. groupName:{}, namesrvAddr:{}", rocketMQConsumerAttribute.getGroupName(), rocketMQConsumerAttribute.getNamesrvAddr());
            return mqConsumer;
        } catch (Exception e) {
            logger.error("RocketMQ consumer started failed", e);
            throw new MQException(e);
        }
    }



    public RocketMQConsumerAttribute buildRocketMQConsumerAttribute(String propertyPrefix, MQZone mqZone) throws MQException {
        RocketMQConsumerAttribute rocketMQConsumerAttribute = new RocketMQConsumerAttribute();

        // 设置基础属性
        rocketMQConsumerAttribute.setNamesrvAddr(getProperty(propertyPrefix + ".namesrvAddr"));
        rocketMQConsumerAttribute.setCanaryNamesrvAddr(getProperty(propertyPrefix + ".canaryNamesrvAddr"));

        // 设置实例名称，支持默认值
        rocketMQConsumerAttribute.setInstanceName(
                getPropertyWithDefault(propertyPrefix + ".instanceName", "rocketmq.default-consumer.instanceName")
        );

        // 设置消费组名称
        rocketMQConsumerAttribute.setGroupName(getProperty(propertyPrefix + ".groupName"));

        // 设置消费线程最小值，支持默认值
        rocketMQConsumerAttribute.setConsumeThreadMin(
                rocketMQConsumerAttribute(propertyPrefix + ".consumeThreadMin", "rocketmq.default-consumer.consumeThreadMin", 0)
        );

        // 设置消费线程最大值，支持默认值
        rocketMQConsumerAttribute.setConsumeThreadMax(
                rocketMQConsumerAttribute(propertyPrefix + ".consumeThreadMax", "rocketmq.default-consumer.consumeThreadMax", 0)
        );

        // 设置 Ack 消息主题（可选）
        Optional.ofNullable(getProperty(propertyPrefix + ".ackMsgTopic"))
                .filter(StringUtils::isNotBlank)
                .ifPresent(rocketMQConsumerAttribute::setAckMsgTopic);

        // 灰度环境的特殊处理
        if (mqZone == MQZone.CANARY) {
            rocketMQConsumerAttribute.setNamesrvAddr(getProperty("rocketmq.canaryNamesrvAddr"));
        }

        return rocketMQConsumerAttribute;
    }

    private DefaultMQPushConsumer createAndInitMQConsumer(RocketMQConsumerAttribute rocketMQConsumerAttribute) throws MQException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketMQConsumerAttribute.getGroupName());
        try {
            // 设置基础属性
            consumer.setNamesrvAddr(rocketMQConsumerAttribute.getNamesrvAddr());
            consumer.setInstanceName(rocketMQConsumerAttribute.getInstanceName());
            consumer.setConsumeThreadMin(rocketMQConsumerAttribute.getConsumeThreadMin());
            consumer.setConsumeThreadMax(rocketMQConsumerAttribute.getConsumeThreadMax());

            // 设置消费模式
            configureMessageModel(consumer, rocketMQConsumerAttribute.getMessageModel());

            // 初始化订阅关系和消息处理器
            ConcurrentMessageListener messageListener = new ConcurrentMessageListener();
            initializeSubscriptions(consumer, messageListener, rocketMQConsumerAttribute);

            // 注册消息监听器
            consumer.registerMessageListener(messageListener);
        } catch (Exception e) {
            logger.error("Failed to initialize RocketMQ consumer, attributes={}", rocketMQConsumerAttribute, e);
            throw new MQException("Failed to initialize RocketMQ consumer", e);
        }

        return consumer;
    }

    private void configureMessageModel(DefaultMQPushConsumer consumer, String messageModel) {
        if (StringUtils.equals(messageModel, MQConstant.CONSUMER.MESSAGE_MODEL.CLUSTER)) {
            consumer.setMessageModel(MessageModel.CLUSTERING);
        } else if (StringUtils.equals(messageModel, MQConstant.CONSUMER.MESSAGE_MODEL.BROADCASTING)) {
            consumer.setMessageModel(MessageModel.BROADCASTING);
        }
    }

    private void initializeSubscriptions(DefaultMQPushConsumer consumer,
                                         ConcurrentMessageListener messageListener,
                                         RocketMQConsumerAttribute rocketMQConsumerAttribute) throws MQException {
        try {
            // 设置普通订阅关系
            for (Map.Entry<Subscription, MQProcessor> entry : rocketMQConsumerAttribute.getSub2Process().entrySet()) {
                Subscription subscription = entry.getKey();
                consumer.subscribe(subscription.getTopic(), subscription.getExpression());
                messageListener.addMQProcessor(subscription.getTopic(), entry.getValue());
            }

            // 设置 ACK 消息订阅关系
            if (!rocketMQConsumerAttribute.getAckMsgTags().isEmpty()) {
                String ackTagsExpression = StringUtils.join(rocketMQConsumerAttribute.getAckMsgTags(), " || ");
                String ackTopic = rocketMQConsumerAttribute.getAckMsgTopic();
                if (StringUtils.isBlank(ackTopic)) {
                    throw new MQException("Ack message topic is null");
                }
                consumer.subscribe(ackTopic, ackTagsExpression);
                messageListener.addMQProcessor(ackTopic, this.ackProcessor);
            }
        } catch (Exception e) {
            throw new MQException("Failed to initialize subscriptions for consumer", e);
        }
    }


    /**
     * 获取属性值，如果为空则返回默认值的配置。
     */
    private String getPropertyWithDefault(String key, String defaultKey) {
        return Optional.ofNullable(getProperty(key))
                .filter(StringUtils::isNotBlank)
                .orElseGet(() -> getProperty(defaultKey));
    }

    /**
     * 获取整数属性值，如果为空则返回默认值的配置。
     */
    private int rocketMQConsumerAttribute(String key, String defaultKey, int defaultValue) {
        String value = getPropertyWithDefault(key, defaultKey);
        return StringUtils.isBlank(value) ? defaultValue : Integer.parseInt(value);
    }

    /**
     * 简单获取属性
     */
    private String getProperty(String key) {
        return environment.getProperty(key);
    }
}
